package org.pentaho.obscommon;

import com.google.common.annotations.VisibleForTesting;
import com.obs.services.exception.ObsException;
import com.obs.services.model.*;
import org.apache.commons.vfs2.*;
import org.apache.commons.vfs2.provider.AbstractFileName;
import org.apache.commons.vfs2.provider.AbstractFileObject;
import org.pentaho.di.core.logging.LogChannel;
import org.pentaho.di.core.logging.LogChannelInterface;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.ArrayList;
import java.util.List;

public abstract class OBSCommonFileObject extends AbstractFileObject {
    private static final Logger logger = LoggerFactory.getLogger(OBSCommonFileObject.class);
    private static LogChannelInterface log = new LogChannel( OBSCommonFileObject.class.getName() );
    public static final String DELIMITER = "/";
    protected OBSCommonFileSystem fileSystem;
    protected String bucketName;
    protected String key;
    protected ObsObject obsObject;
    protected ObjectMetadata obsObjectMetadata;
    protected AbstractFileName fileName;
    protected String empty;


    protected OBSCommonFileObject( AbstractFileName name, OBSCommonFileSystem fileSystem ) throws Exception {
        super( name, fileSystem );
        this.fileName = name;
        this.fileSystem = fileSystem;
        this.bucketName = this.getOBSBucketName();
        this.key = this.getBucketRelativeOBSPath();
    }

    protected long doGetContentSize() {
        return this.obsObjectMetadata.getContentLength();
    }

    protected InputStream doGetInputStream() throws Exception {
        logger.debug( "Accessing content {}", this.getQualifiedName() );
        this.closeOBSObject();
        ObsObject streamOBSObject = this.getOBSObject();
        return new OBSCommonFileInputStream(streamOBSObject.getObjectContent(), streamOBSObject);
    }


    @Override public void createFile() throws FileSystemException {
        //PDI-19598: Copied from super.createFile() but it was a way to force the file creation on OBS
        synchronized ( fileSystem ) {
            try {
                // VFS-210: We do not want to trunc any existing file, checking for its existence is
                // still required
                if ( exists() && !isFile() ) {
                    throw new FileSystemException( "vfs.provider/create-file.error", super.getName() );
                }

                if ( !exists() ) {
                    OutputStream outputStream = getOutputStream();
                    //Force to write an empty array to force file creation on OBS bucket
                    outputStream.write( new byte[] {} );
                    outputStream.close();
                    endOutput();
                }
            } catch ( final RuntimeException re ) {
                throw re;
            } catch ( final Exception e ) {
                throw new FileSystemException( "vfs.provider/create-file.error", super.getName(), e );
            }
        }
    }

    protected FileType doGetType() throws Exception {
        FileType fileType = this.getType();
        return fileType;
    }

    protected String[] doListChildren() throws Exception {
        List<String> childrenList = new ArrayList();

        // only listing folders or the root bucket
        if (this.getType() == FileType.FOLDER || this.isRootBucket()) {
            childrenList = this.getOBSObjectsFromVirtualFolder(this.key, this.bucketName);
        }
        String[] childrenArr = new String[((List)childrenList).size()];

        return (String[])((List)childrenList).toArray(childrenArr);
    }

    protected String getOBSBucketName() {
        String bucket = this.getName().getPath();
        if (bucket.indexOf(DELIMITER, 1) > 1) {
            bucket = bucket.substring(1, bucket.indexOf(DELIMITER, 1));
        } else {
            bucket = bucket.replace(DELIMITER, "");
        }
        return bucket;
    }

    protected List<String> getOBSObjectsFromVirtualFolder(String key, String bucketName) throws Exception {
        List<String> childrenList = new ArrayList();

        // fix cases where the path doesn't include the final delimiter
        String realKey = key;
        if ( !key.endsWith( DELIMITER ) ) {
            realKey = key + DELIMITER;
        }

        if ("".equals(key) && "".equals(bucketName)) {
            //Getting buckets in root folder
            ListBucketsRequest listBucketsRequest = new ListBucketsRequest();
            List<ObsBucket> bucketList = this.fileSystem.getOBSClient().listBuckets(listBucketsRequest);
            for ( ObsBucket bucket : bucketList ) {
                childrenList.add( bucket.getBucketName() + DELIMITER );
            }
        } else {
            this.getObjectsFromNonRootFolder( key, bucketName, childrenList, realKey );
        }
        return childrenList;
    }

    private void getObjectsFromNonRootFolder(String key, String bucketName, List<String> childrenList, String realKey) {
        String prefix = key.isEmpty() || key.endsWith( DELIMITER ) ? key : key + DELIMITER;

        ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
        listObjectsRequest.setBucketName( bucketName );
        listObjectsRequest.setPrefix( prefix );
        listObjectsRequest.setDelimiter( DELIMITER );

        ObjectListing ol = this.fileSystem.getOBSClient().listObjects( listObjectsRequest );

        ArrayList<ObsObject> allSummaries = new ArrayList<>( ol.getObjects() );
        ArrayList<String> allCommonPrefixes = new ArrayList<>( ol.getCommonPrefixes() );

        // get full list
        while(ol.isTruncated()) {
            ol = this.fileSystem.getOBSClient().listObjects( String.valueOf(ol) );
            allSummaries.addAll(ol.getObjects());
            allCommonPrefixes.addAll(ol.getCommonPrefixes());
        }

        for ( ObsObject obsos : allSummaries ) {
            if ( !obsos.getObjectKey().equals( realKey ) ) {
                childrenList.add( obsos.getObjectKey().substring( prefix.length() ) );
            }
        }

        for ( String commonPrefix : allCommonPrefixes ) {
            if ( !commonPrefix.equals( realKey ) ) {
                childrenList.add( commonPrefix.substring( prefix.length() ) );
            }
        }
    }

    protected String getBucketRelativeOBSPath() {
        if ( getName().getPath().indexOf( DELIMITER, 1 ) >= 0 ) {
            return getName().getPath().substring( getName().getPath().indexOf( DELIMITER, 1 ) + 1 );
        } else {
            return "";
        }
    }

    @VisibleForTesting
    public ObsObject getOBSObject() {
        return this.getOBSObject( this.key, this.bucketName );
    }

    protected ObsObject getOBSObject(String key, String bucket) {
        if (this.obsObject != null && this.obsObject.getObjectContent() != null) {
            logger.debug("Returning exisiting object {}", this.getQualifiedName());
            return this.obsObject;
        } else {
            logger.debug("Getting object {}", this.getQualifiedName());
            return this.fileSystem.getOBSClient().getObject(bucket, key);
        }
    }

    protected boolean isRootBucket() {
        return this.key.equals("");
    }

    public void doAttach() throws Exception {
        logger.debug("Attach called on {}", this.getQualifiedName());
        this.injectType(FileType.IMAGINARY);

        if (this.isRootBucket()) {
            this.injectType(FileType.FOLDER);
            return;
        }
        try {
            // 1. Is it an existing file?
            this.obsObjectMetadata = this.fileSystem.getOBSClient().getObjectMetadata(this.bucketName, this.key);
            this.injectType(this.getName().getType());
        } catch (ObsException var5) {
            // 2. Is it in reality a folder?
            this.handleAttachException(this.key, this.bucketName);
        } finally {
            this.closeOBSObject();
        }
    }

    protected void handleAttachException(String key, String bucket) throws IOException {
        String keyWithDelimiter = key + DELIMITER;

        try {
            this.obsObjectMetadata = this.fileSystem.getOBSClient().getObjectMetadata(this.bucketName, key);
            this.injectType(FileType.FOLDER);
            this.key = keyWithDelimiter;
        } catch (ObsException var12) {
            int responseCode = var12.getResponseCode();
            String responseStatus = var12.getResponseStatus();

            try {
                if (responseCode == 404 && responseStatus.equals("Not Found")) {
                    this.obsObject = this.getOBSObject(keyWithDelimiter, bucket);
                    this.obsObjectMetadata = this.obsObject.getMetadata();
                    this.injectType(FileType.FOLDER);
                    this.key = keyWithDelimiter;
                } else {
                    this.handleAttachExceptionFallback(bucket, keyWithDelimiter, var12);
                }
            } catch (ObsException var11) {
                this.handleAttachExceptionFallback(bucket, keyWithDelimiter, var11);
            }
        } finally {
            this.closeOBSObject();
        }

    }

    private void handleAttachExceptionFallback(String bucket, String keyWithDelimiter, ObsException exception) throws FileSystemException {

        ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
        listObjectsRequest.setBucketName(bucket);
        listObjectsRequest.setPrefix(keyWithDelimiter);
        listObjectsRequest.setDelimiter(DELIMITER);
        ObjectListing ol = this.fileSystem.getOBSClient().listObjects(listObjectsRequest);

        if ( !( ol.getCommonPrefixes().isEmpty() && ol.getObjects().isEmpty() ) ) {
            injectType( FileType.FOLDER );
        } else {
            //Folders don't really exist - they will generate a "NoSuchKey" exception
            // confirms key doesn't exist but connection okay
            String errorCode = exception.getErrorCode();
            if ( !errorCode.equals( "NoSuchKey" ) ) {
                // bubbling up other connection errors
                logger.error( "Could not get information on " + getQualifiedName(),
                        exception ); // make sure this gets printed for the user
                throw new FileSystemException( "vfs.provider/get-type.error", getQualifiedName(), exception );
            }
        }
    }

    private void closeOBSObject() throws IOException {
        if (this.obsObject != null) {
            InputStream is = this.obsObject.getObjectContent();
            if (is != null){
                is.close();
            }
            this.obsObject = null;
        }

    }

    public void doDetach() throws Exception {
        logger.debug("detaching {}", this.getQualifiedName());
        this.closeOBSObject();
    }

    protected void doDelete() throws FileSystemException {
        this.doDelete(this.key, this.bucketName);
    }

    protected void doDelete(String key, String bucketName) throws FileSystemException {
        // can only delete folder if empty
        if (this.getType() == FileType.FOLDER) {
            // list all children inside the folder
            ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
            listObjectsRequest.setBucketName(this.bucketName);
            listObjectsRequest.setPrefix(this.key);
            ObjectListing ol = this.fileSystem.getOBSClient().listObjects(listObjectsRequest);
            ArrayList<ObsObject> allSummaries = new ArrayList(ol.getObjects());

            // get full list
            while(ol.isTruncated()) {
                ol = this.fileSystem.getOBSClient().listObjects(String.valueOf(ol));
                allSummaries.addAll(ol.getObjects());
            }

            for ( ObsObject obsos : allSummaries ) {
                fileSystem.getOBSClient().deleteObject( bucketName, obsos.getObjectKey() );
            }
        }
        this.fileSystem.getOBSClient().deleteObject(bucketName, key);
    }

    protected OutputStream doGetOutputStream(boolean bAppend) throws Exception {

        if (obsObjectMetadata != null) {
            if (this.getOBSObject().getMetadata().getContentLength() == 0){
                bAppend = false;
            }
            if (bAppend) {
                InputStream objectContent = this.doGetInputStream();
                InputStreamReader inputStreamReader = new InputStreamReader(objectContent);
                Long length = obsObjectMetadata.getContentLength();
                char[] chars = new char[length.intValue()];
                int len = -1;
                while ((len = inputStreamReader.read(chars)) != -1) {
                    this.empty = new String(chars, 0, len);
                }
                objectContent.close();
            }
        } else {
            this.empty = "";
        }
        return new OBSCommonPipedOutputStream(this.fileSystem, this.bucketName, this.key, bAppend, this.empty);
    }

    @Override
    public long doGetLastModifiedTime() {
        if ( obsObjectMetadata != null && obsObjectMetadata.getLastModified() != null ) {
            return obsObjectMetadata.getLastModified().getTime();
        } else {
            // In some case obs system might not return modified time.
            logger.info( "No last modified date is available for this object" );
            return 0L;
        }
    }


    @Override
    protected void doCreateFolder() throws Exception {
        if ( !this.isRootBucket() ) {
            // create meta-data for your folder and set content-length to 0
            ObjectMetadata metadata = new ObjectMetadata();
            metadata.setContentLength(0L);
            metadata.setContentType("binary/octet-stream");

            // create empty content
            InputStream emptyContent = new ByteArrayInputStream(new byte[0]);

            // create a PutObjectRequest passing the folder name suffixed by /
            PutObjectRequest putObjectRequest = this.createPutObjectRequest(this.bucketName, this.key + DELIMITER, emptyContent, metadata);

            // send request to OBS to create folder
            try {
                this.fileSystem.getOBSClient().putObject(putObjectRequest);
            } catch (ObsException var5) {
                throw new FileSystemException("vfs.provider.local/create-folder.error", this, var5);
            }
        } else {
            throw new FileSystemException("vfs.provider/create-folder-not-supported.error");
        }
    }

    protected PutObjectRequest createPutObjectRequest(String bucketName, String key, InputStream inputStream, ObjectMetadata objectMetadata) {
        PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, key, inputStream);
        putObjectRequest.setMetadata(objectMetadata);
        return putObjectRequest;
    }

    @Override
    public void moveTo(final FileObject destFile) throws FileSystemException {
        if (canRenameTo(destFile)) {
            if (!getParent().isWriteable()) {
                throw new FileSystemException("vfs.provider/rename-parent-read-only.error", getName(),
                        getParent().getName());
            }
        } else {
            if (!isWriteable()) {
                throw new FileSystemException("vfs.provider/rename-read-only.error", getName());
            }
        }

        if (destFile.exists() && !isSameFile(destFile)) {
            destFile.deleteAll();
            // throw new FileSystemException("vfs.provider/rename-dest-exists.error", destFile.getName());
        }

        if (canRenameTo(destFile)) {
            // issue rename on same filesystem
            try {
                // remember type to avoid attach
                doRename(destFile);
                destFile.close(); // now the destFile is no longer imaginary. force reattach.
                handleDelete(); // fire delete-events. This file-object (src) is like deleted.
            } catch (final RuntimeException re) {
                throw re;
            } catch (final Exception exc) {
                throw new FileSystemException("vfs.provider/rename.error", exc, getName(), destFile.getName());
            }
        } else {
            // different fs - do the copy/delete stuff
            destFile.copyFrom(this, Selectors.SELECT_SELF);

            if ((destFile.getType().hasContent()
                    && destFile.getFileSystem().hasCapability(Capability.SET_LAST_MODIFIED_FILE)
                    || destFile.getType().hasChildren()
                    && destFile.getFileSystem().hasCapability(Capability.SET_LAST_MODIFIED_FOLDER))
                    && fileSystem.hasCapability(Capability.GET_LAST_MODIFIED)) {
                destFile.getContent().setLastModifiedTime(this.getContent().getLastModifiedTime());
            }
        }

    }

    protected void doRename(FileObject newFile) throws Exception {
        if (this.getType().equals(FileType.FOLDER)) {
            throw new FileSystemException("vfs.provider/rename-not-supported.error");
        } else {
            this.obsObjectMetadata = this.fileSystem.getOBSClient().getObjectMetadata(this.bucketName, this.key);
            if (this.obsObjectMetadata == null) {
                throw new FileSystemException("vfs.provider/rename.error", new Object[]{this, newFile});
            } else {
                OBSCommonFileObject dest = (OBSCommonFileObject)newFile;
                CopyObjectRequest copyObjRequest = this.createCopyObjectRequest(this.bucketName, this.key, dest.bucketName, dest.key);
                this.fileSystem.getOBSClient().copyObject(copyObjRequest);
                this.delete();
            }
        }
    }

    protected CopyObjectRequest createCopyObjectRequest(String sourceBucket, String sourceKey, String destBucket, String destKey) {
        return new CopyObjectRequest(sourceBucket, sourceKey, destBucket, destKey);
    }

    protected String getQualifiedName() {
        return this.getQualifiedName(this);
    }

    protected String getQualifiedName(OBSCommonFileObject obsFileObject) {
        return obsFileObject.bucketName + DELIMITER + obsFileObject.key;
    }
}
